#pragma once

#include <map>
#include <mutex>

#include <Core/Types.h>
#include <IO/OpenedFile.h>
#include <Common/ProfileEvents.h>


#include <city.h>


namespace ProfileEvents
{
extern const Event OpenedFileCacheHits;
extern const Event OpenedFileCacheMisses;
extern const Event OpenedFileCacheMicroseconds;
}

namespace DB
{
/** Cache of opened files for reading.
  * It allows to share file descriptors when doing reading with 'pread' syscalls on readonly files.
  * Note: open/close of files is very cheap on Linux and we should not bother doing it 10 000 times a second.
  * (This may not be the case on Windows with WSL. This is also not the case if strace is active. Neither when some eBPF is loaded).
  * But sometimes we may end up opening one file multiple times, that increases chance exhausting opened files limit.
  */
class OpenedFileCache
{
    class OpenedFileMap
    {
        using Key = std::pair<std::string /* path */, int /* flags */>;

        using OpenedFileWeakPtr = std::weak_ptr<OpenedFile>;
        using Files = std::map<Key, OpenedFileWeakPtr>;

        Files files;
        std::mutex mutex;

    public:
        using OpenedFilePtr = std::shared_ptr<OpenedFile>;

        OpenedFilePtr get(const std::string & path, int flags)
        {
            Key key(path, flags);

            std::lock_guard lock(mutex);

            auto [it, inserted] = files.emplace(key, OpenedFilePtr{});
            if (!inserted)
            {
                if (auto res = it->second.lock())
                {
                    ProfileEvents::increment(ProfileEvents::OpenedFileCacheHits);
                    return res;
                }
            }
            ProfileEvents::increment(ProfileEvents::OpenedFileCacheMisses);

            OpenedFilePtr res{new OpenedFile(path, flags), [key, this](auto ptr) {
                                  {
                                      std::lock_guard another_lock(mutex);
                                      files.erase(key);
                                  }
                                  delete ptr;
                              }};

            it->second = res;
            return res;
        }

        void remove(const std::string & path, int flags)
        {
            Key key(path, flags);
            std::lock_guard lock(mutex);
            files.erase(key);
        }
    };

    static constexpr size_t buckets = 1024;
    std::vector<OpenedFileMap> impls{buckets};

public:
    using OpenedFilePtr = OpenedFileMap::OpenedFilePtr;

    OpenedFilePtr get(const std::string & path, int flags)
    {
        const auto bucket = hash(path);
        return impls[bucket].get(path, flags);
    }

    void remove(const std::string & path, int flags)
    {
        const auto bucket = hash(path);
        impls[bucket].remove(path, flags);
    }

    static OpenedFileCache & instance()
    {
        static OpenedFileCache res;
        return res;
    }

private:
    static UInt64 hash(const std::string & path)
    {
        return CityHash_v1_0_2::CityHash64(path.data(), path.length()) % buckets;
    }

};

using OpenedFileCachePtr = std::shared_ptr<OpenedFileCache>;
}
